/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.rocketmq.client.impl.factory;

import java.io.UnsupportedEncodingException;
import java.net.DatagramSocket;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;

import com.alibaba.rocketmq.client.ClientConfig;
import com.alibaba.rocketmq.client.admin.MQAdminExtInner;
import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.impl.ClientRemotingProcessor;
import com.alibaba.rocketmq.client.impl.FindBrokerResult;
import com.alibaba.rocketmq.client.impl.MQAdminImpl;
import com.alibaba.rocketmq.client.impl.MQClientAPIImpl;
import com.alibaba.rocketmq.client.impl.MQClientManager;
import com.alibaba.rocketmq.client.impl.consumer.DefaultMQPullConsumerImpl;
import com.alibaba.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl;
import com.alibaba.rocketmq.client.impl.consumer.MQConsumerInner;
import com.alibaba.rocketmq.client.impl.consumer.ProcessQueue;
import com.alibaba.rocketmq.client.impl.consumer.PullMessageService;
import com.alibaba.rocketmq.client.impl.consumer.RebalanceService;
import com.alibaba.rocketmq.client.impl.producer.DefaultMQProducerImpl;
import com.alibaba.rocketmq.client.impl.producer.MQProducerInner;
import com.alibaba.rocketmq.client.impl.producer.TopicPublishInfo;
import com.alibaba.rocketmq.client.log.ClientLogger;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.stat.ConsumerStatsManager;
import com.alibaba.rocketmq.common.MQVersion;
import com.alibaba.rocketmq.common.MixAll;
import com.alibaba.rocketmq.common.ServiceState;
import com.alibaba.rocketmq.common.UtilAll;
import com.alibaba.rocketmq.common.conflict.PackageConflictDetect;
import com.alibaba.rocketmq.common.constant.PermName;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.message.MessageQueue;
import com.alibaba.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
import com.alibaba.rocketmq.common.protocol.body.ConsumerRunningInfo;
import com.alibaba.rocketmq.common.protocol.heartbeat.ConsumeType;
import com.alibaba.rocketmq.common.protocol.heartbeat.ConsumerData;
import com.alibaba.rocketmq.common.protocol.heartbeat.HeartbeatData;
import com.alibaba.rocketmq.common.protocol.heartbeat.ProducerData;
import com.alibaba.rocketmq.common.protocol.heartbeat.SubscriptionData;
import com.alibaba.rocketmq.common.protocol.route.BrokerData;
import com.alibaba.rocketmq.common.protocol.route.QueueData;
import com.alibaba.rocketmq.common.protocol.route.TopicRouteData;
import com.alibaba.rocketmq.remoting.RPCHook;
import com.alibaba.rocketmq.remoting.common.RemotingHelper;
import com.alibaba.rocketmq.remoting.exception.RemotingException;
import com.alibaba.rocketmq.remoting.netty.NettyClientConfig;

/**
 * @author shijia.wxr<vintage.wang@gmail.com>
 * @since 2013-6-15
 */
public class MQClientInstance {
	private final static long LockTimeoutMillis = 3000;
	private final Logger log = ClientLogger.getLog();
	private final ClientConfig clientConfig;
	private final int instanceIndex;
	private final String clientId;
	private final long bootTimestamp = System.currentTimeMillis();
	private final ConcurrentHashMap<String/* group */, MQProducerInner> producerTable = new ConcurrentHashMap<String, MQProducerInner>();
	private final ConcurrentHashMap<String/* group */, MQConsumerInner> consumerTable = new ConcurrentHashMap<String, MQConsumerInner>();
	private final ConcurrentHashMap<String/* group */, MQAdminExtInner> adminExtTable = new ConcurrentHashMap<String, MQAdminExtInner>();
	private final NettyClientConfig nettyClientConfig;
	private final MQClientAPIImpl mQClientAPIImpl;
	private final MQAdminImpl mQAdminImpl;
	private final ConcurrentHashMap<String/* Topic */, TopicRouteData> topicRouteTable = new ConcurrentHashMap<String, TopicRouteData>();
	private final Lock lockNamesrv = new ReentrantLock();
	private final Lock lockHeartbeat = new ReentrantLock();
	private final ConcurrentHashMap<String/* Broker Name */, HashMap<Long/* brokerId */, String/* address */>> brokerAddrTable = new ConcurrentHashMap<String, HashMap<Long, String>>();
	private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
		@Override
		public Thread newThread(Runnable r) {
			return new Thread(r, "MQClientFactoryScheduledThread");
		}
	});
	private final ClientRemotingProcessor clientRemotingProcessor;
	private final PullMessageService pullMessageService;
	private final RebalanceService rebalanceService;
	private final DefaultMQProducer defaultMQProducer;
	private ServiceState serviceState = ServiceState.CREATE_JUST;
	private DatagramSocket datagramSocket;

	private final ConsumerStatsManager consumerStatsManager;

	public MQClientInstance(ClientConfig clientConfig, int instanceIndex, String clientId, RPCHook rpcHook) {
		this.clientConfig = clientConfig;
		this.instanceIndex = instanceIndex;
		this.nettyClientConfig = new NettyClientConfig();
		this.nettyClientConfig.setClientCallbackExecutorThreads(clientConfig.getClientCallbackExecutorThreads());
		// 6.1）初始化ClientRemotingProcessor对象，处理接受的事件请求；
		this.clientRemotingProcessor = new ClientRemotingProcessor(this);
		// 6.2）初始化MQClientAPIImpl对象
		// 在初始化过程中，初始化了MQClientAPIImpl.remotingClient:NettyRemotingClient对象，
		// 将ClientRemotingProcessor对象作为事件处理器注册到NettyRemotingClient对象中，
		// 处理的事件号有：CHECK_TRANSACTION_STATE、NOTIFY_CONSUMER_IDS_CHANGED、RESET_CONSUMER_CLIENT_OFFSET、GET_CONSUMER_STATUS_FROM_CLIENT、GET_CONSUMER_RUNNING_INFO、CONSUME_MESSAGE_DIRECTLY。
		// ============即执行ClientRemotingProcessor的processRequest方法处理相应的事件=======================
		this.mQClientAPIImpl = new MQClientAPIImpl(this.nettyClientConfig, this.clientRemotingProcessor, rpcHook);

		// 6.3）若ClientConfig.namesrvAddr不为空，则拆分成数组存入MQClientAPIImpl.remotingClient变量中；
		if (this.clientConfig.getNamesrvAddr() != null) {
			this.mQClientAPIImpl.updateNameServerAddressList(this.clientConfig.getNamesrvAddr());
			log.info("user specified name server address: {}", this.clientConfig.getNamesrvAddr());
		}

		this.clientId = clientId;

		this.mQAdminImpl = new MQAdminImpl(this);
		// 6.4.1）初始化PullMessageService服务线程供DefaultMQPushConsumer端使用
		this.pullMessageService = new PullMessageService(this);

		// 6.4.2）初始化RebalanceService服务线程供Consumser端使用
		this.rebalanceService = new RebalanceService(this);

		// 6.5）初始化producerGroup等于"CLIENT_INNER_PRODUCER"的DefaultMQProducer对象；
		this.defaultMQProducer = new DefaultMQProducer(MixAll.CLIENT_INNER_PRODUCER_GROUP);
		this.defaultMQProducer.resetClientConfig(clientConfig);

		// 6.4.3）初始化ConsumerStatsManager服务线程
		this.consumerStatsManager = new ConsumerStatsManager(this.scheduledExecutorService);

		log.info("created a new client Instance, FactoryIndex: {} ClinetID: {} {} {}", //
				this.instanceIndex, //
				this.clientId, //
				this.clientConfig, //
				MQVersion.getVersionDesc(MQVersion.CurrentVersion));
	}

	public MQClientInstance(ClientConfig clientConfig, int instanceIndex, String clientId) {
		this(clientConfig, instanceIndex, clientId, null);
	}

	public void start() throws MQClientException {
		PackageConflictDetect.detectFastjson();

		synchronized (this) {
			// 1、检查MQClientInstance.ServiceState的状态（初始化状态为ServiceState.CREATE_JUST）；
			// 只有状态为CREATE_JUST时才启动该Producer；其他状态均不执行启动过程；
			switch (this.serviceState) {
				case CREATE_JUST :
					// 2、将MQClientInstance的ServiceState置为start_failed,以免重复启动；
					this.serviceState = ServiceState.START_FAILED;

					// If not specified,looking address from name server
					// 3、检查是否设置了NameServer的地址（在启动Producer之前在应用层设置ClientConfig.namesrvAddr）
					if (null == this.clientConfig.getNamesrvAddr()) {
						/**
						 * 若没有则从地址服务器找Name Server地址，
						 * 地址服务器的路径是"http://&quot; +WS_DOMAIN_NAME+":8080/rocketmq/"+WS_DOMAIN_SUBGROUP；
						 * 其中WS_DOMAIN_NAME为system.property配置参数rocketmq.namesrv.domain的值，默认jmenv.tbsite.net
						 * WS_DOMAIN_SUBG为system.property配置参数rocketmq.namesrv.domain.subgroup的值，默认nsaddr
						 */
						this.clientConfig.setNamesrvAddr(this.mQClientAPIImpl.fetchNameServerAddr());
					}

					// **********************************
					// 4、启动nettyClient端远程通信
					// 接受并处理Broker返回的响应消息
					// 当发送拉取消息在Broker返回响应消息之后调用NettyRemotingAbstract.processMessageReceived(ChannelHandlerContext ctx, RemotingCommand msg)方法
					this.mQClientAPIImpl.start();

					// 5、启动各类定时任务
					this.startScheduledTask();

					// **************************************************
					// 6、启动PullMessageService服务线程，
					// 该服务监听PullMessageService.pullRequestQueue：LinkedBlockingQueue<PullRequest>队列，
					// 若该队列有拉取消息的请求，则选择Consumer进行消息的拉取，该定时服务是Consumer使用的；
					this.pullMessageService.start();

					// ****************************************************
					// 7、启动消费端负载均衡服务RebalanceService 消费端会通过RebalanceService线程，10秒钟做一次基于topic下的所有队列负载,该服务是Consumer端使用的；
					this.rebalanceService.start();

					// Start push service
					// 8、启动在初始化MQClientInstance对象过程中初始化的DefaultMQProducer对象，
					// 即调用内部初始化的DefaultMQProducer对象的DefaultMQProducerImpl.start（false）方法，
					// 参数为false表示在启动该Producer对象的内部不启动MQClientInstance；
					this.defaultMQProducer.getDefaultMQProducerImpl().start(false);
					log.info("the client factory [{}] start OK", this.clientId);

					// 9、设置MQClientInstance的ServiceState为RUNNING；
					this.serviceState = ServiceState.RUNNING;
					break;
				case RUNNING :
					break;
				case SHUTDOWN_ALREADY :
					break;
				case START_FAILED :
					throw new MQClientException("The Factory object[" + this.getClientId() + "] has been created before, and failed.", null);
				default :
					break;
			}
		}
	}

	private void startScheduledTask() {

		if (null == this.clientConfig.getNamesrvAddr()) {

			this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

				@Override
				public void run() {
					try {
						// 定时获取nameserver地址
						// 5.1）若Name Server地址（ClientConfig.namesrvAddr）为空，
						// 则设置定时获取Name Server地址，每隔120秒从地址服务器查找Name Server地址；
						MQClientInstance.this.mQClientAPIImpl.fetchNameServerAddr();
					} catch (Exception e) {
						log.error("ScheduledTask fetchNameServerAddr exception", e);
					}
				}
			}, 1000 * 10, 1000 * 60 * 2, TimeUnit.MILLISECONDS);
		}

		this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

			@Override
			public void run() {
				try {
					// 定时从nameserver获取topic路由信息
					// 5.2）每隔30秒调用一次MQClientInstance.updateTopicRouteInfoFromNameServer()方法更新topic的路由信息；
					MQClientInstance.this.updateTopicRouteInfoFromNameServer();
				} catch (Exception e) {
					log.error("ScheduledTask updateTopicRouteInfoFromNameServer exception", e);
				}
			}
		}, 10, this.clientConfig.getPollNameServerInteval(), TimeUnit.MILLISECONDS);

		this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

			@Override
			public void run() {
				try {
					// 定时清理下线的borker
					// 5.3）每隔30秒进行一次清理离线的Broker的工作。
					// 遍历MQClientInstance.brokerAddrTable的值，对于不在MQClientInstance.topicRouteTable中的Broker地址，从brokerAddrTable中清理掉；
					MQClientInstance.this.cleanOfflineBroker();

					// ****************定时向所有broker发送心跳信息，（包括订阅关系）*************
					// 5.4）每隔30秒调用一次MQClientInstance.sendHeartbeatToAllBrokerWithLock()方法，在该方法中，
					// 第一，向所有在MQClientInstance.brokerAddrTable列表中的Broker发送心跳消息；
					// 第二，向Filter过滤服务器发送REGISTER_MESSAGE_FILTER_CLASS请求码，更新过滤服务器中的Filterclass文件；
					MQClientInstance.this.sendHeartbeatToAllBrokerWithLock();
				} catch (Exception e) {
					log.error("ScheduledTask sendHeartbeatToAllBroker exception", e);
				}
			}
		}, 1000, this.clientConfig.getHeartbeatBrokerInterval(), TimeUnit.MILLISECONDS);

		this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

			@Override
			public void run() {
				try {
					// 定时持久化Consumer消费进度（广播存储到本地，集群存储到Broker）
					// 5.5）每隔5秒调用一次MQClientInstance.persistAllConsumerOffset()方法。
					// 将每个Consumer端的消费进度信息向Broker同步（详细步骤见4.2.2小节），
					// 只有Consumer端才会进行同步；Producer端不发起同步；
					MQClientInstance.this.persistAllConsumerOffset();
				} catch (Exception e) {
					log.error("ScheduledTask persistAllConsumerOffset exception", e);
				}
			}
		}, 1000 * 10, this.clientConfig.getPersistConsumerOffsetInterval(), TimeUnit.MILLISECONDS);

		this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

			@Override
			public void run() {
				try {
					// 动态调整消费线程池
					// 5.6）定期每1毫秒执行一次MQClientInstance.adjustThreadPool()方法检测并调整PUSH模式
					// （DefaultMQPushConsumer）下ConsumeMessageService对象中线程池的线程数）
					MQClientInstance.this.adjustThreadPool();
				} catch (Exception e) {
					log.error("ScheduledTask adjustThreadPool exception", e);
				}
			}
		}, 1, 1, TimeUnit.MINUTES);
	}

	/**
	 * Remove offline broker
	 */
	private void cleanOfflineBroker() {
		try {
			if (this.lockNamesrv.tryLock(LockTimeoutMillis, TimeUnit.MILLISECONDS))
				try {
					ConcurrentHashMap<String, HashMap<Long, String>> updatedTable = new ConcurrentHashMap<String, HashMap<Long, String>>();

					Iterator<Entry<String, HashMap<Long, String>>> itBrokerTable = this.brokerAddrTable.entrySet().iterator();
					while (itBrokerTable.hasNext()) {
						Entry<String, HashMap<Long, String>> entry = itBrokerTable.next();
						String brokerName = entry.getKey();
						HashMap<Long, String> oneTable = entry.getValue();

						HashMap<Long, String> cloneAddrTable = new HashMap<Long, String>();
						cloneAddrTable.putAll(oneTable);

						Iterator<Entry<Long, String>> it = cloneAddrTable.entrySet().iterator();
						while (it.hasNext()) {
							Entry<Long, String> ee = it.next();
							String addr = ee.getValue();
							if (!this.isBrokerAddrExistInTopicRouteTable(addr)) {
								it.remove();
								log.info("the broker addr[{} {}] is offline, remove it", brokerName, addr);
							}
						}

						if (cloneAddrTable.isEmpty()) {
							itBrokerTable.remove();
							log.info("the broker[{}] name's host is offline, remove it", brokerName);
						} else {
							updatedTable.put(brokerName, cloneAddrTable);
						}
					}

					if (!updatedTable.isEmpty()) {
						this.brokerAddrTable.putAll(updatedTable);
					}
				} finally {
					this.lockNamesrv.unlock();
				}
		} catch (InterruptedException e) {
			log.warn("cleanOfflineBroker Exception", e);
		}
	}

	private boolean isBrokerAddrExistInTopicRouteTable(final String addr) {
		Iterator<Entry<String, TopicRouteData>> it = this.topicRouteTable.entrySet().iterator();
		while (it.hasNext()) {
			Entry<String, TopicRouteData> entry = it.next();
			TopicRouteData topicRouteData = entry.getValue();
			List<BrokerData> bds = topicRouteData.getBrokerDatas();
			for (BrokerData bd : bds) {
				if (bd.getBrokerAddrs() != null) {
					boolean exist = bd.getBrokerAddrs().containsValue(addr);
					if (exist)
						return true;
				}
			}
		}

		return false;
	}

	private void persistAllConsumerOffset() {
		Iterator<Entry<String, MQConsumerInner>> it = this.consumerTable.entrySet().iterator(); // ConcurrentHashMap<String/* group */, MQConsumerInner>
		while (it.hasNext()) {
			Entry<String, MQConsumerInner> entry = it.next();
			MQConsumerInner impl = entry.getValue();
			impl.persistConsumerOffset();
		}
	}

	public void sendHeartbeatToAllBrokerWithLock() {
		if (this.lockHeartbeat.tryLock()) {
			try {
				// 向所有在MQClientInstance.brokerAddrTable列表中的Broker发送心跳消息
				this.sendHeartbeatToAllBroker();
				// 向Filter过滤服务器发送REGISTER_MESSAGE_FILTER_CLASS请求码，更新过滤服务器中的Filterclass文件
				this.uploadFilterClassSource();
			} catch (final Exception e) {
				log.error("sendHeartbeatToAllBroker exception", e);
			} finally {
				this.lockHeartbeat.unlock();
			}
		} else {
			log.warn("lock heartBeat, but failed.");
		}
	}

	private void uploadFilterClassToAllFilterServer(final String consumerGroup, final String fullClassName, final String topic, final String filterClassSource) throws UnsupportedEncodingException {
		byte[] classBody = null;
		int classCRC = 0;
		try {
			classBody = filterClassSource.getBytes(MixAll.DEFAULT_CHARSET);
			classCRC = UtilAll.crc32(classBody);
		} catch (Exception e1) {
			log.warn("uploadFilterClassToAllFilterServer Exception, ClassName: {} {}", //
					fullClassName, //
					RemotingHelper.exceptionSimpleDesc(e1));
		}

		TopicRouteData topicRouteData = this.topicRouteTable.get(topic);
		if (topicRouteData != null //
				&& topicRouteData.getFilterServerTable() != null && !topicRouteData.getFilterServerTable().isEmpty()) {
			Iterator<Entry<String, List<String>>> it = topicRouteData.getFilterServerTable().entrySet().iterator();
			while (it.hasNext()) {
				Entry<String, List<String>> next = it.next();
				List<String> value = next.getValue();
				for (final String fsAddr : value) {
					try {
						this.mQClientAPIImpl.registerMessageFilterClass(fsAddr, consumerGroup, topic, fullClassName, classCRC, classBody, 5000);

						log.info("register message class filter to {} OK, ConsumerGroup: {} Topic: {} ClassName: {}", fsAddr, consumerGroup, topic, fullClassName);

					} catch (Exception e) {
						log.error("uploadFilterClassToAllFilterServer Exception", e);
					}
				}
			}
		} else {
			log.warn("register message class filter failed, because no filter server, ConsumerGroup: {} Topic: {} ClassName: {}", consumerGroup, topic, fullClassName);
		}
	}

	private void uploadFilterClassSource() {
		Iterator<Entry<String, MQConsumerInner>> it = this.consumerTable.entrySet().iterator();
		while (it.hasNext()) {
			Entry<String, MQConsumerInner> next = it.next();
			MQConsumerInner consumer = next.getValue();
			if (ConsumeType.CONSUME_PASSIVELY == consumer.consumeType()) {
				Set<SubscriptionData> subscriptions = consumer.subscriptions();
				for (SubscriptionData sub : subscriptions) {
					if (sub.isClassFilterMode() && sub.getFilterClassSource() != null) {
						final String consumerGroup = consumer.groupName();
						final String className = sub.getSubString();
						final String topic = sub.getTopic();
						final String filterClassSource = sub.getFilterClassSource();
						try {
							this.uploadFilterClassToAllFilterServer(consumerGroup, className, topic, filterClassSource);
						} catch (Exception e) {
							log.error("uploadFilterClassToAllFilterServer Exception", e);
						}
					}
				}
			}
		}
	}

	private void sendHeartbeatToAllBroker() {
		// 1、初始化HeartbeatData对象
		final HeartbeatData heartbeatData = this.prepareHeartbeatData();
		final boolean producerEmpty = heartbeatData.getProducerDataSet().isEmpty();
		final boolean consumerEmpty = heartbeatData.getConsumerDataSet().isEmpty();
		// 4、若ConsumerData集合和ProducerData集合都为空，说明没有consumer或produer，则不发送心跳信息；
		if (producerEmpty && consumerEmpty) {
			log.warn("sending hearbeat, but no consumer and no producer");
			return;
		}

		// 5、若ConsumerData集合和ProducerData集合不是都为空，
		// 则遍历MQClientInstance.brokerAddrTable列表，向每个Broker地址发送请求码为HEART_BEAT的心跳消息，
		// 但是当存在Consumer时才向所有Broker发送心跳消息，
		// 否则若不存在Consumer则只向主用Broker地址发送心跳消息；
		Iterator<Entry<String, HashMap<Long, String>>> it = this.brokerAddrTable.entrySet().iterator();
		while (it.hasNext()) {
			Entry<String, HashMap<Long, String>> entry = it.next();
			String brokerName = entry.getKey();
			HashMap<Long, String> oneTable = entry.getValue();
			if (oneTable != null) {
				for (Long id : oneTable.keySet()) {
					String addr = oneTable.get(id);
					if (addr != null) {
						if (consumerEmpty) {
							// 若不存在Consumer则只向主用Broker地址发送心跳消息
							if (id != MixAll.MASTER_ID)
								continue;
						}

						try {
							this.mQClientAPIImpl.sendHearbeat(addr, heartbeatData, 3000);
							log.info("send heart beat to broker[{} {} {}] success", brokerName, id, addr);
							log.info(heartbeatData.toString());
						} catch (Exception e) {
							log.error("send heart beat to broker exception", e);
						}
					}
				}
			}
		}
	}

	private HeartbeatData prepareHeartbeatData() {
		// 初始化HeartbeatData对象
		HeartbeatData heartbeatData = new HeartbeatData();

		// clientID
		// 1、将该Producer或Consumer的ClientID赋值给HeartbeatData对象的clientID变量
		heartbeatData.setClientID(this.clientId);

		// Consumer
		for (String group : this.consumerTable.keySet()) {
			// 2、遍历MQClientInstance.consumerTable: ConcurrentHashMap<String/* group */, MQConsumerInner>变量，
			// 根据每个MQConsumerInner对象的值初始化ConsumerData对象
			// （包括GroupName、consumeType、MessageModel、consumeFromWhere、rebalanceImpl变量中的RebalanceImpl. subscriptionInner值初始化subscriptionDataSet:Set<SubscriptionData>变量、UnitMode）
			MQConsumerInner impl = this.consumerTable.get(group);
			if (impl != null) {
				ConsumerData consumerData = new ConsumerData();
				consumerData.setGroupName(impl.groupName());
				consumerData.setConsumeType(impl.consumeType());
				consumerData.setMessageModel(impl.messageModel());
				consumerData.setConsumeFromWhere(impl.consumeFromWhere());
				consumerData.getSubscriptionDataSet().addAll(impl.subscriptions());
				consumerData.setUnitMode(impl.isUnitMode());

				// 将该consumerData对象放入HeartbeatData对象的ConsumerData集合变量consumerDataSet:Set<ConsumerData>中；
				heartbeatData.getConsumerDataSet().add(consumerData);
			}
		}

		// Producer
		for (String group : this.producerTable.keySet()) {
			// 3、遍历MQClientInstance.producerTable: ConcurrentHashMap<String/* group */, MQProducerInner>变量；
			// 根据每条记录的ProducerGroup值初始化ProducerData对象，
			MQProducerInner impl = this.producerTable.get(group);
			if (impl != null) {
				ProducerData producerData = new ProducerData();
				producerData.setGroupName(group);

				// 放入HeartbeatData对象的ProducerData集合变量producerDataSet:Set<ProducerData>中；
				heartbeatData.getProducerDataSet().add(producerData);
			}
		}

		return heartbeatData;
	}

	public void updateTopicRouteInfoFromNameServer() {
		Set<String> topicList = new HashSet<String>();

		// Consumer
		{
			Iterator<Entry<String, MQConsumerInner>> it = this.consumerTable.entrySet().iterator();
			while (it.hasNext()) {
				Entry<String, MQConsumerInner> entry = it.next();
				MQConsumerInner impl = entry.getValue();
				if (impl != null) {
					Set<SubscriptionData> subList = impl.subscriptions();
					if (subList != null) {
						for (SubscriptionData subData : subList) {
							topicList.add(subData.getTopic());
						}
					}
				}
			}
		}

		// Producer
		{
			Iterator<Entry<String, MQProducerInner>> it = this.producerTable.entrySet().iterator();
			while (it.hasNext()) {
				Entry<String, MQProducerInner> entry = it.next();
				MQProducerInner impl = entry.getValue();
				if (impl != null) {
					Set<String> lst = impl.getPublishTopicList();
					topicList.addAll(lst);
				}
			}
		}

		for (String topic : topicList) {
			this.updateTopicRouteInfoFromNameServer(topic);
		}
	}

	public boolean updateTopicRouteInfoFromNameServer(final String topic) {
		return updateTopicRouteInfoFromNameServer(topic, false, null);
	}

	public boolean updateTopicRouteInfoFromNameServer(final String topic, boolean isDefault, DefaultMQProducer defaultMQProducer) {
		try {
			if (this.lockNamesrv.tryLock(LockTimeoutMillis, TimeUnit.MILLISECONDS)) {
				try {
					// private String orderTopicConf;
					// private List<QueueData> queueDatas; ---------- brokerName;readQueueNums;writeQueueNums;perm;topicSynFlag
					// private List<BrokerData> brokerDatas;
					// private HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;
					TopicRouteData topicRouteData;
					if (isDefault && defaultMQProducer != null) {
						/**
						 * 2.1）isDefault=true并且defaultMQProducer不等于null，
						 * 则以主题名"TBW102"为topic向NameServer发送GET_ROUTEINTO_BY_TOPIC 请求码的请求从NameServer获取对应的Broker信息和topic配置信息，
						 * 即返回TopicRouteData对象。
						 */
						topicRouteData = this.mQClientAPIImpl.getDefaultTopicRouteInfoFromNameServer(defaultMQProducer.getCreateTopicKey(), 1000 * 3);
						if (topicRouteData != null) {
							for (QueueData data : topicRouteData.getQueueDatas()) {
								int queueNums = Math.min(defaultMQProducer.getDefaultTopicQueueNums(), data.getReadQueueNums());
								data.setReadQueueNums(queueNums);
								data.setWriteQueueNums(queueNums);
							}
						}
					} else {
						/**
						 * 2.1）isDefault=false，defaultMQProducer=null，
						 * 则以参数topic向NameServer发送GET_ROUTEINTO_BY_TOPIC 请求码的请求从NameServer获取对应的Broker信息和topic配置信息，
						 * 即返回TopicRouteData对象
						 */
						topicRouteData = this.mQClientAPIImpl.getTopicRouteInfoFromNameServer(topic, 1000 * 3);
					}
					// 2.2）若第2.1步获取的TopicRouteData对象为空则直接返回，否则继续下面的操作
					if (topicRouteData != null) {
						// 2.2）根据topic参数值从MQClientInstance.topicRouteTable本地列表中获取TopicRouteData对象；
						TopicRouteData old = this.topicRouteTable.get(topic);
						// 2.3）比较从本地列表获取的TopicRouteData对象与从NameServer获取的TopicRouteData对象是否相等
						boolean changed = topicRouteDataIsChange(old, topicRouteData);
						if (!changed) {
							/**
							 * 本地列表获取的TopicRouteData对象与从NameServer获取的TopicRouteData对象若相等
							 * 则再检查MQClientInstance.consumerTable
							 * 或者producerTable变量中每个Producer或Consumer端是否都包含该topic的订阅信息
							 * 或者订阅信息中的MessageQueue队列是否为空，
							 * 若没有订阅信息或者订阅信息中的队列为空则返回true并继续执行2.4步的操作；
							 */
							changed = this.isNeedUpdateTopicRouteInfo(topic);
						} else {
							log.info("the topic[{}] route info changed, odl[{}] ,new[{}]", topic, old, topicRouteData);
						}

						if (changed) {
							TopicRouteData cloneTopicRouteData = topicRouteData.cloneTopicRouteData();

							// 2.4遍历从NameServer获取的TopicRouteData对象的BrokerDatas变量
							for (BrokerData bd : topicRouteData.getBrokerDatas()) {
								// 将每个BrokerData对象的BrokerName和BrokerAddr:HashMap<Long/*brokerId */, String/* broker address*/>为key、values值存入 MQClientInstance.brokerAddrTable: ConcurrentHashMap<String/*BrokerName*/,HashMap<Long/*brokerId*/, String/*address*/>>变量中 ；
								this.brokerAddrTable.put(bd.getBrokerName(), bd.getBrokerAddrs());
							}

							// Update Pub info
							// 2.5）更新Producer端的topic信息
							{
								// A)创建TopicPublishInfo对象
								TopicPublishInfo publishInfo = topicRouteData2TopicPublishInfo(topic, topicRouteData);
								// B）设置TopicPublishInfo.haveTopicRouterInfo等于true，表示已经有最新topic信息，即TopicPublishInfo对象；
								publishInfo.setHaveTopicRouterInfo(true);
								// 遍历MQClientInstance.producerTable: ConcurrentHashMap<String/* group */, MQProducerInner>变量
								Iterator<Entry<String, MQProducerInner>> it = this.producerTable.entrySet().iterator();
								while (it.hasNext()) {
									Entry<String, MQProducerInner> entry = it.next();
									MQProducerInner impl = entry.getValue();
									if (impl != null) {
										// 调用每个DefaultMQProducerImpl对象的updateTopicPublishInfo(String topic, TopicPublishInfo info)方法，
										// 以该topic为key值、上一步返回的TopicPublishInfo对象为values值 存入该DefaultMQProducerImpl.topicPublishInfoTable变量中；
										impl.updateTopicPublishInfo(topic, publishInfo);
									}
								}
							}

							// Update sub info
							// 2.6）更新Consumer端的topic信息：
							{
								Set<MessageQueue> subscribeInfo = topicRouteData2TopicSubscribeInfo(topic, topicRouteData);
								Iterator<Entry<String, MQConsumerInner>> it = this.consumerTable.entrySet().iterator();
								while (it.hasNext()) {
									Entry<String, MQConsumerInner> entry = it.next();
									// 遍历MQClientInstance.consumerTable:ConcurrentHashMap<String/* group */, MQConsumerInner>中的MQConsumerInner对象；
									MQConsumerInner impl = entry.getValue();
									if (impl != null) {
										impl.updateTopicSubscribeInfo(topic, subscribeInfo);
									}
								}
							}
							log.info("topicRouteTable.put TopicRouteData[{}]", cloneTopicRouteData);
							// 2.7）将该topic以及TopicRouteData对象存入MQClientInstance.topicRouteTable:ConcurrentHashMap<String/* Topic */,TopicRouteData>变量中；
							this.topicRouteTable.put(topic, cloneTopicRouteData);
							return true;
						}
					} else {
						log.warn("updateTopicRouteInfoFromNameServer, getTopicRouteInfoFromNameServer return null, Topic: {}", topic);
					}
				} catch (Exception e) {
					if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX) && !topic.equals(MixAll.DEFAULT_TOPIC)) {
						log.warn("updateTopicRouteInfoFromNameServer Exception", e);
					}
				} finally {
					this.lockNamesrv.unlock();
				}
			} else {
				log.warn("updateTopicRouteInfoFromNameServer tryLock timeout {}ms", LockTimeoutMillis);
			}
		} catch (InterruptedException e) {
			log.warn("updateTopicRouteInfoFromNameServer Exception", e);
		}

		return false;
	}

	private boolean topicRouteDataIsChange(TopicRouteData olddata, TopicRouteData nowdata) {
		if (olddata == null || nowdata == null)
			return true;
		TopicRouteData old = olddata.cloneTopicRouteData();
		TopicRouteData now = nowdata.cloneTopicRouteData();
		Collections.sort(old.getQueueDatas());
		Collections.sort(old.getBrokerDatas());
		Collections.sort(now.getQueueDatas());
		Collections.sort(now.getBrokerDatas());
		return !old.equals(now);

	}

	public static TopicPublishInfo topicRouteData2TopicPublishInfo(final String topic, final TopicRouteData route) {
		TopicPublishInfo info = new TopicPublishInfo();
		// A.1）若TopicRouteData对象的orderTopicConf不为空并且长度大于零
		if (route.getOrderTopicConf() != null && route.getOrderTopicConf().length() > 0) {
			// 则以";"解析成数组
			String[] brokers = route.getOrderTopicConf().split(";");
			// brokers数组的每个成员是以":"分隔的，结构是"brokerName:queueNum"；
			for (String broker : brokers) {
				String[] item = broker.split(":");
				int nums = Integer.parseInt(item[1]);
				// 遍历数组的每个成员变量，为每个成员变量创建queueNum个MessageQueue对象
				for (int i = 0; i < nums; i++) {
					MessageQueue mq = new MessageQueue(topic, item[0], i);
					// 存入TopicPublishInfo.messageQueueList: List<MessageQueue>变量中
					info.getMessageQueueList().add(mq);
				}
			}
			// 遍历结束之后将TopicPublishInfo.orderTopic设置为true；
			info.setOrderTopic(true);
		} else {
			// A.2）若TopicRouteData对象的orderTopicConf为空，则取TopicRouteData对象的QueueData列表；然后遍历该QueueData列表的每个QueueData对象
			List<QueueData> qds = route.getQueueDatas();
			Collections.sort(qds);
			for (QueueData qd : qds) {
				// A.2.1）检查该QueueData对象是否具有写权限；若没有则继续遍历下一个QueueData对象；
				if (PermName.isWriteable(qd.getPerm())) {
					BrokerData brokerData = null;
					for (BrokerData bd : route.getBrokerDatas()) {
						// 对于QueueData对象qd的BrokerName值，若该值等于TopicRouteData对象的BrokerData列表中某一BrokerData对象的BrokerName值
						if (bd.getBrokerName().equals(qd.getBrokerName())) {
							brokerData = bd;
							break;
						}
					}

					if (null == brokerData) {
						continue;
					}
					// 再检查BrokerData对象的BrokerAddr:HashMap<Long/* brokerId */, String/* broker address */>列表中是否包含了主用Broker（ID=0）,
					// 即检查BrokerName名下是否有主用Broker;
					if (!brokerData.getBrokerAddrs().containsKey(MixAll.MASTER_ID)) {
						// 若该BrokerName集群下面没有主用Broker，则不创建写入队列MessageQueue对象；
						// A.2.3）继续遍历下一个QueueData对象,直到遍历完该QueueData列表为止
						continue;
					}

					// 若存在主用Broker，
					// 则取该QueueData对象的writeQueueNums值，
					for (int i = 0; i < qd.getWriteQueueNums(); i++) {
						// 以该Broker的name、入参topic、从0开始的序号（小于writeQueueNums）为参数
						// 初始化writeQueueNums个MessageQueue对象，
						MessageQueue mq = new MessageQueue(topic, qd.getBrokerName(), i);
						// 存入TopicPublishInfo.messageQueueList:List<MessageQueue>变量中
						info.getMessageQueueList().add(mq);
					}
				}
			}
			// 最后将TopicPublishInfo.orderTopic设置为false；
			info.setOrderTopic(false);
		}

		return info;
	}

	public static Set<MessageQueue> topicRouteData2TopicSubscribeInfo(final String topic, final TopicRouteData route) {
		Set<MessageQueue> mqList = new HashSet<MessageQueue>();
		List<QueueData> qds = route.getQueueDatas();
		// 遍历TopicRouteData对象的QueueData列表中每个QueueData对象
		for (QueueData qd : qds) {
			// 首先判断该QueueData对象是否具有读权限，
			if (PermName.isReadable(qd.getPerm())) {
				// 若有则根据该QueueData对象的readQueueNums值，
				for (int i = 0; i < qd.getReadQueueNums(); i++) {
					// 创建readQueueNums个MessageQueue对象
					MessageQueue mq = new MessageQueue(topic, qd.getBrokerName(), i);
					// 并构成MessageQueue集合
					mqList.add(mq);
				}
			}
		}

		return mqList;
	}

	private boolean isNeedUpdateTopicRouteInfo(final String topic) {
		boolean result = false;
		{
			// 对于Producer端
			Iterator<Entry<String, MQProducerInner>> it = this.producerTable.entrySet().iterator();
			while (it.hasNext() && !result) {
				Entry<String, MQProducerInner> entry = it.next();
				MQProducerInner impl = entry.getValue();
				if (impl != null) {
					result = impl.isPublishTopicNeedUpdate(topic);
				}
			}
		}

		{
			// 对于PushConsumer、PullConsumer端
			Iterator<Entry<String, MQConsumerInner>> it = this.consumerTable.entrySet().iterator();
			while (it.hasNext() && !result) {
				Entry<String, MQConsumerInner> entry = it.next();
				MQConsumerInner impl = entry.getValue();
				if (impl != null) {
					result = impl.isSubscribeTopicNeedUpdate(topic);
				}
			}
		}

		return result;
	}

	public void shutdown() {
		// Consumer
		if (!this.consumerTable.isEmpty())
			return;

		// AdminExt
		if (!this.adminExtTable.isEmpty())
			return;

		// Producer
		if (this.producerTable.size() > 1)
			return;

		synchronized (this) {
			switch (this.serviceState) {
				case CREATE_JUST :
					break;
				case RUNNING :
					this.defaultMQProducer.getDefaultMQProducerImpl().shutdown(false);

					this.serviceState = ServiceState.SHUTDOWN_ALREADY;
					this.pullMessageService.shutdown(true);
					this.scheduledExecutorService.shutdown();
					this.mQClientAPIImpl.shutdown();
					this.rebalanceService.shutdown();

					if (this.datagramSocket != null) {
						this.datagramSocket.close();
						this.datagramSocket = null;
					}
					MQClientManager.getInstance().removeClientFactory(this.clientId);
					log.info("the client factory [{}] shutdown OK", this.clientId);
					break;
				case SHUTDOWN_ALREADY :
					break;
				default :
					break;
			}
		}
	}

	public boolean registerConsumer(final String group, final MQConsumerInner consumer) {
		if (null == group || null == consumer) {
			return false;
		}
		// 若在该变量中已存在该consumerGroup的记录则向应用层抛出MQClientException异常
		// 说明在一个客户端的一个进程下面启动多个Consumer时consumerGroup名字不能一样，否则无法启动；
		MQConsumerInner prev = this.consumerTable.putIfAbsent(group, consumer); // consumerTable===>ConcurrentHashMap<String/* group */, MQConsumerInner>
		if (prev != null) {
			log.warn("the consumer group[" + group + "] exist already.");
			return false;
		}

		return true;
	}

	public void unregisterConsumer(final String group) {
		this.consumerTable.remove(group);
		this.unregisterClientWithLock(null, group);
	}

	private void unregisterClientWithLock(final String producerGroup, final String consumerGroup) {
		try {
			if (this.lockHeartbeat.tryLock(LockTimeoutMillis, TimeUnit.MILLISECONDS)) {
				try {
					this.unregisterClient(producerGroup, consumerGroup);
				} catch (Exception e) {
					log.error("unregisterClient exception", e);
				} finally {
					this.lockHeartbeat.unlock();
				}
			} else {
				log.warn("lock heartBeat, but failed.");
			}
		} catch (InterruptedException e) {
			log.warn("unregisterClientWithLock exception", e);
		}
	}

	private void unregisterClient(final String producerGroup, final String consumerGroup) {
		Iterator<Entry<String, HashMap<Long, String>>> it = this.brokerAddrTable.entrySet().iterator();
		while (it.hasNext()) {
			Entry<String, HashMap<Long, String>> entry = it.next();
			String brokerName = entry.getKey();
			HashMap<Long, String> oneTable = entry.getValue();

			if (oneTable != null) {
				for (Long id : oneTable.keySet()) {
					String addr = oneTable.get(id);
					if (addr != null) {
						try {
							this.mQClientAPIImpl.unregisterClient(addr, this.clientId, producerGroup, consumerGroup, 3000);
							log.info("unregister client[Producer: {} Consumer: {}] from broker[{} {} {}] success", producerGroup, consumerGroup, brokerName, id, addr);
						} catch (RemotingException e) {
							log.error("unregister client exception from broker: " + addr, e);
						} catch (MQBrokerException e) {
							log.error("unregister client exception from broker: " + addr, e);
						} catch (InterruptedException e) {
							log.error("unregister client exception from broker: " + addr, e);
						}
					}
				}
			}
		}
	}

	public boolean registerProducer(final String group, final DefaultMQProducerImpl producer) {
		if (null == group || null == producer) {
			return false;
		}

		MQProducerInner prev = this.producerTable.putIfAbsent(group, producer);
		if (prev != null) {
			log.warn("the producer group[{}] exist already.", group);
			return false;
		}

		return true;
	}

	public void unregisterProducer(final String group) {
		this.producerTable.remove(group);
		this.unregisterClientWithLock(group, null);
	}

	public boolean registerAdminExt(final String group, final MQAdminExtInner admin) {
		if (null == group || null == admin) {
			return false;
		}

		MQAdminExtInner prev = this.adminExtTable.putIfAbsent(group, admin);
		if (prev != null) {
			log.warn("the admin group[{}] exist already.", group);
			return false;
		}

		return true;
	}

	public void unregisterAdminExt(final String group) {
		this.adminExtTable.remove(group);
	}

	public void rebalanceImmediately() {
		this.rebalanceService.wakeup();
	}

	public void doRebalance() {
		// 每隔10秒钟就执行一次该方法，
		// 在该方法中遍历MQClientInstance.consumerTable:ConcurrentHashMap<String/* group */, MQConsumerInner>变量
		for (String group : this.consumerTable.keySet()) {
			MQConsumerInner impl = this.consumerTable.get(group);
			if (impl != null) {
				try {
					// push
					// pull
					// ******************************************
					impl.doRebalance();
				} catch (Exception e) {
					log.error("doRebalance exception", e);
				}
			}
		}
	}

	public MQProducerInner selectProducer(final String group) {
		return this.producerTable.get(group);
	}

	public MQConsumerInner selectConsumer(final String group) {
		return this.consumerTable.get(group);
	}

	public FindBrokerResult findBrokerAddressInAdmin(final String brokerName) {
		String brokerAddr = null;
		boolean slave = false;
		boolean found = false;

		HashMap<Long/* brokerId */, String/* address */> map = this.brokerAddrTable.get(brokerName);
		if (map != null && !map.isEmpty()) {
			FOR_SEG : for (Map.Entry<Long, String> entry : map.entrySet()) {
				Long id = entry.getKey();
				brokerAddr = entry.getValue();
				if (brokerAddr != null) {
					found = true;
					if (MixAll.MASTER_ID == id) {
						slave = false;
						break FOR_SEG;
					} else {
						slave = true;
					}
					break;

				}
			} // end of for
		}

		if (found) {
			return new FindBrokerResult(brokerAddr, slave);
		}

		return null;
	}

	public String findBrokerAddressInPublish(final String brokerName) {
		HashMap<Long/* brokerId */, String/* address */> map = this.brokerAddrTable.get(brokerName);
		if (map != null && !map.isEmpty()) {
			return map.get(MixAll.MASTER_ID);
		}

		return null;
	}

	public FindBrokerResult findBrokerAddressInSubscribe(//
			final String brokerName, //
			final long brokerId, //
			final boolean onlyThisBroker//
	) {
		String brokerAddr = null;
		boolean slave = false;
		boolean found = false;

		HashMap<Long/* brokerId */, String/* address */> map = this.brokerAddrTable.get(brokerName);
		if (map != null && !map.isEmpty()) {
			// 根据brokerName和brokerId参数从MQClientInstance.brokerAddrTable: ConcurrentHashMap<String/* Broker Name */, HashMap<Long/* brokerId */, String/* address */>>变量中获取对应的Broker地址
			brokerAddr = map.get(brokerId);
			slave = (brokerId != MixAll.MASTER_ID);
			found = (brokerAddr != null);

			if (!found && !onlyThisBroker) {
				Entry<Long, String> entry = map.entrySet().iterator().next();
				// 若获取不到则从brokerName下面的Map列表中找其他地址返回即可；
				brokerAddr = entry.getValue();
				slave = (entry.getKey() != MixAll.MASTER_ID);
				found = true;
			}
		}

		if (found) {
			return new FindBrokerResult(brokerAddr, slave);
		}

		return null;
	}

	public List<String> findConsumerIdList(final String topic, final String group) {
		// 2、获取topic的Broker地址
		String brokerAddr = this.findBrokerAddrByTopic(topic);
		if (null == brokerAddr) {
			// 3、若上一步获取的Broker地址为空，
			// 则调用MQClientInstance. updateTopicRouteInfoFromNameServer方法从NameServer获取之后再按第2步的方法查找topic对应的Broker地址；
			// 循环执行直到找到Broker地址为止；
			this.updateTopicRouteInfoFromNameServer(topic);
			brokerAddr = this.findBrokerAddrByTopic(topic);
		}

		if (null != brokerAddr) {
			try {
				// 选择一台broker获取基于group的所有消费端（有心跳向所有broker注册客户端信息）
				/**
				 * 4、以consumerGroup为参数向该Broker发送GET_CONSUMER_LIST_BY_GROUP请求码，获取以该consumerGroup为名的ClientId集合；
				 * 在Broker收到请求之后，从ConsumerManager.consumerTable中获取ConsumerGroupInfo对象，
				 * 然后遍历该对象的channelInfoTable集合，将集合的每个ClientChannelInfo对象的clientId变量值作为ClientId集合；
				 */
				return this.mQClientAPIImpl.getConsumerIdListByGroup(brokerAddr, group, 3000);
			} catch (Exception e) {
				log.warn("getConsumerIdListByGroup exception, " + brokerAddr + " " + group, e);
			}
		}

		return null;
	}

	public String findBrokerAddrByTopic(final String topic) {
		// 以入参topic值从MQClientInstance.topicRouteTable:ConcurrentHashMap<String/* Topic*/,TopicRouteData>变量中获取对应的TopicRouteData对象，
		TopicRouteData topicRouteData = this.topicRouteTable.get(topic);
		if (topicRouteData != null) {
			// 取该TopicRouteData对象的brokerDatas:List<BrokerData>集合
			List<BrokerData> brokers = topicRouteData.getBrokerDatas();
			if (!brokers.isEmpty()) {
				// 若该集合不为空，则取第一个BrokerData对象
				BrokerData bd = brokers.get(0);
				// 从该对象的brokerAddrs: HashMap<Long/* brokerId */, String/* broker address */>变量中获取brokerId=0的Broker地址（即主用Broker），若没有主用Broker则获取备用Broker地址；
				return bd.selectBrokerAddr();
			}
		}
		// 该集合为空则直接返回null
		return null;
	}

	public void resetOffset(String topic, String group, Map<MessageQueue, Long> offsetTable) {
		DefaultMQPushConsumerImpl consumer = null;
		try {
			MQConsumerInner impl = this.consumerTable.get(group);
			if (impl != null && impl instanceof DefaultMQPushConsumerImpl) {
				consumer = (DefaultMQPushConsumerImpl) impl;
			} else {
				log.info("[reset-offset] consumer dose not exist. group={}", group);
				return;
			}

			ConcurrentHashMap<MessageQueue, ProcessQueue> processQueueTable = consumer.getRebalanceImpl().getProcessQueueTable();
			Iterator<MessageQueue> itr = processQueueTable.keySet().iterator();
			while (itr.hasNext()) {
				MessageQueue mq = itr.next();
				if (topic.equals(mq.getTopic())) {
					ProcessQueue pq = processQueueTable.get(mq);
					pq.setDropped(true);
					pq.clear();
				}
			}

			Iterator<MessageQueue> iterator = offsetTable.keySet().iterator();
			while (iterator.hasNext()) {
				MessageQueue mq = iterator.next();
				consumer.updateConsumeOffset(mq, offsetTable.get(mq));
				log.info("[reset-offset] reset offsetTable. topic={}, group={}, mq={}, offset={}", new Object[]{topic, group, mq, offsetTable.get(mq)});
			}
			consumer.getOffsetStore().persistAll(offsetTable.keySet());

			try {
				TimeUnit.SECONDS.sleep(10);
			} catch (InterruptedException e) {
				//
			}

			iterator = offsetTable.keySet().iterator();
			while (iterator.hasNext()) {
				MessageQueue mq = iterator.next();
				consumer.updateConsumeOffset(mq, offsetTable.get(mq));
				log.info("[reset-offset] reset offsetTable. topic={}, group={}, mq={}, offset={}", new Object[]{topic, group, mq, offsetTable.get(mq)});
			}
			consumer.getOffsetStore().persistAll(offsetTable.keySet());

			iterator = offsetTable.keySet().iterator();
			processQueueTable = consumer.getRebalanceImpl().getProcessQueueTable();
			while (iterator.hasNext()) {
				MessageQueue mq = iterator.next();
				processQueueTable.remove(mq);
			}
		} finally {
			consumer.getRebalanceImpl().doRebalance();
		}
	}

	public Map<MessageQueue, Long> getConsumerStatus(String topic, String group) {
		MQConsumerInner impl = this.consumerTable.get(group);
		if (impl != null && impl instanceof DefaultMQPushConsumerImpl) {
			DefaultMQPushConsumerImpl consumer = (DefaultMQPushConsumerImpl) impl;
			return consumer.getOffsetStore().cloneOffsetTable(topic);
		} else if (impl != null && impl instanceof DefaultMQPullConsumerImpl) {
			DefaultMQPullConsumerImpl consumer = (DefaultMQPullConsumerImpl) impl;
			return consumer.getOffsetStore().cloneOffsetTable(topic);
		} else {
			return Collections.EMPTY_MAP;
		}
	}

	public TopicRouteData getAnExistTopicRouteData(final String topic) {
		return this.topicRouteTable.get(topic);
	}

	public MQClientAPIImpl getMQClientAPIImpl() {
		return mQClientAPIImpl;
	}

	public MQAdminImpl getMQAdminImpl() {
		return mQAdminImpl;
	}

	public String getClientId() {
		return clientId;
	}

	public long getBootTimestamp() {
		return bootTimestamp;
	}

	public ScheduledExecutorService getScheduledExecutorService() {
		return scheduledExecutorService;
	}

	public PullMessageService getPullMessageService() {
		return pullMessageService;
	}

	public DefaultMQProducer getDefaultMQProducer() {
		return defaultMQProducer;
	}

	public ConcurrentHashMap<String, TopicRouteData> getTopicRouteTable() {
		return topicRouteTable;
	}

	public void adjustThreadPool() {
		// 遍历MQClientInstance.consumerTable:ConcurrentHashMap<String/* group */, MQConsumerInner>变量
		Iterator<Entry<String, MQConsumerInner>> it = this.consumerTable.entrySet().iterator();
		while (it.hasNext()) {
			Entry<String, MQConsumerInner> entry = it.next();
			MQConsumerInner impl = entry.getValue();
			if (impl != null) {
				try {
					if (impl instanceof DefaultMQPushConsumerImpl) {
						DefaultMQPushConsumerImpl dmq = (DefaultMQPushConsumerImpl) impl;
						// 若MQConsumerInner为DefaultMQPushConsumerImpl，则调用DefaultMQPushConsumerImpl.adjustThreadPool()方法
						dmq.adjustThreadPool();
					}
				} catch (Exception e) {
				}
			}
		}
	}

	public ConsumeMessageDirectlyResult consumeMessageDirectly(final MessageExt msg, //
			final String consumerGroup, //
			final String brokerName) {
		MQConsumerInner mqConsumerInner = this.consumerTable.get(consumerGroup);
		if (null != mqConsumerInner) {
			DefaultMQPushConsumerImpl consumer = (DefaultMQPushConsumerImpl) mqConsumerInner;

			ConsumeMessageDirectlyResult result = consumer.getConsumeMessageService().consumeMessageDirectly(msg, brokerName);
			return result;
		}

		return null;
	}

	public ConsumerRunningInfo consumerRunningInfo(final String consumerGroup) {
		MQConsumerInner mqConsumerInner = this.consumerTable.get(consumerGroup);

		ConsumerRunningInfo consumerRunningInfo = mqConsumerInner.consumerRunningInfo();

		List<String> nsList = this.mQClientAPIImpl.getRemotingClient().getNameServerAddressList();
		String nsAddr = "";
		if (nsList != null) {
			for (String addr : nsList) {
				nsAddr = nsAddr + addr + ";";
			}
		}

		consumerRunningInfo.getProperties().put(ConsumerRunningInfo.PROP_NAMESERVER_ADDR, nsAddr);
		consumerRunningInfo.getProperties().put(ConsumerRunningInfo.PROP_CONSUME_TYPE, mqConsumerInner.consumeType());
		consumerRunningInfo.getProperties().put(ConsumerRunningInfo.PROP_CLIENT_VERSION, MQVersion.getVersionDesc(MQVersion.CurrentVersion));

		return consumerRunningInfo;
	}

	public ConsumerStatsManager getConsumerStatsManager() {
		return consumerStatsManager;
	}
}
